package oktx.dubbo.terminator;

import com.alibaba.dubbo.common.utils.ReflectUtils;
import com.alibaba.dubbo.registry.support.ConsumerInvokerWrapper;
import com.alibaba.dubbo.registry.support.ProviderConsumerRegTable;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.RpcInvocation;
import oktx.dubbo.participant.DubboTccParticipant;
import oktx.participant.JtaParticipant;
import oktx.participant.Participant;
import oktx.RemoteType;
import oktx.terminator.TransactionTerminatorBean;
import oktx.terminator.Terminator;
import org.springframework.beans.factory.InitializingBean;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

/**
 * Created by huangdachao on 2018/5/16 14:39.
 * 提交或回滚Dubbo远程子事务
 */
public class DubboTxTerminator implements Terminator, InitializingBean {
    private Method commitMethod;
    private Method rollbackMethod;

    @Override
    public void afterPropertiesSet() {
        Method[] methods = TransactionTerminatorBean.class.getMethods();
        for (Method m : methods) {
            if (m.getName().equals("rollback")) {
                rollbackMethod = m;
            } else if (m.getName().equals("commit")) {
                commitMethod = m;
            }
        }
    }

    @Override
    public boolean support(Participant p) {
        return p instanceof DubboTccParticipant || (p instanceof JtaParticipant && p.getType() == RemoteType.dubbo);
    }

    @Override
    public void commit(Participant p) {
        if (p instanceof JtaParticipant) {
            terminate(p, commitMethod);
        } else {
            DubboTccParticipant dp = (DubboTccParticipant) p;
            terminate(p, getTccMethod(dp.getClazz(), dp.getConfirmMethod()));
        }
    }

    @Override
    public void rollback(Participant p) {
        if (p instanceof JtaParticipant) {
            terminate(p, rollbackMethod);
        } else {
            DubboTccParticipant dp = (DubboTccParticipant) p;
            terminate(p, getTccMethod(dp.getClazz(), dp.getCancelMethod()));
        }
    }

    @SuppressWarnings("unchecked")
    private void terminate(Participant participant, Method method) {
        Invocation invocation = new RpcInvocation(method, new Object[]{participant.getXid()});
        List<Invoker> list = new ArrayList<>();
        Set<ConsumerInvokerWrapper> invokers = ProviderConsumerRegTable.getConsumerInvoker(method.getDeclaringClass().getName());
        for (ConsumerInvokerWrapper w : invokers) {
            list.addAll(w.getRegistryDirectory().list(invocation));
        }

        Invoker invoker = list.stream().filter(i -> i.getUrl().getPort() == participant.getPort() && i.getUrl().getHost().equals(participant.getIp()))
                .findFirst().orElseThrow(() -> new RuntimeException("远程服务没找到: " + method.getDeclaringClass().getName() + "." + method.getName()));
        try {
            invoker.invoke(invocation).recreate();
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
    }

    private Method getTccMethod(Class<?> clazz, String methodName) {
        try {
            return ReflectUtils.findMethodByMethodSignature(clazz, methodName, new String[]{String.class.getName()});
        } catch (Exception e) {
            throw new RuntimeException("远程调用方法" + clazz.getName() + "." + methodName + "未定义");
        }
    }

}
